package fnic.prehand.esagent.passitive_measure;

 
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.Response;
import fnic.prehand.augment.Flow;
import fnic.prehand.common.AgentConfig;
import fnic.river.RedisRiver;

public class PassitiveRedisRiver extends RedisRiver implements Runnable{
	private static final AgentConfig config = AgentConfig.getInstance();
	private static final Logger logger = Logger.getLogger(PassitiveRedisRiver.class);
	
	private static volatile boolean runfg = true;
	private static final int FLOWFIELDSIZE = 19;
	private static final int bulksize = config.getRedisoncedealnum();
	private static final String redis_key = config.getPredisKey();
	

	
	private int sleepSec = 4;
//	private static final int NMLSLEEP=4;
	private static final int MINSLEEP=config.getPreadRedis_min();
	private static final int MAXSLEEP=config.getPreadRedis_max();
	
//	private PassitiveAgent esagent;
	
	private Aggregate aggrate ;
	private MakeAugment augmentmaker; //扩维
	private ThreadPoolExecutor scheduler;
	
	private boolean first = true;
	private long record_starttime = 0;
	private long record_time =0;
	private long last_time = 0;
	private long now_time = 0;
	private int aggr_interval = config.getPaggr_interval();
	
	private volatile long docounter = 0;
	private long totalcounter = 0;
	private long cycle = 0;
	
//	PassitiveAgent2 pagentthread;
	ObjectPool objPool;
	
	public PassitiveRedisRiver(String redishost, int redisport, String key){
//		esagent = new PassitiveAgent(config.getPesIndexName(), config.getPesIndexType());
//		pagentthread = new PassitiveAgent2(config.getPesIndexName(),config.getPesIndexType());
//		esindexer = new WorkerThread();
		super(redishost, redisport, key);
        objPool = new ObjectPool();
        objPool.createPool();  
        
//		scheduler = (ThreadPoolExecutor)Executors.newCachedThreadPool();
		scheduler = (ThreadPoolExecutor)Executors.newFixedThreadPool(Math.max(32,Runtime.getRuntime().availableProcessors()));
		aggrate = new Aggregate();
		augmentmaker = new MakeAugment();
	}
	
//	public void bulkGets(){
//		while(runfg){
//			for(int i=0; i<bulksize; i++){
//				rsps[i] = (Response<String>)jpline.rpop(redis_key); 
//			}
//			try{
//				jpline.sync();
//			}catch(Exception e){
//				logger.error("redis sync get rsp fail: " +e.getMessage());
//				continue;
//			}
//			for(int i=0; i<bulksize; i++){
//				if(rsps[i]!=null && rsps[i].get()!=null && rsps[i].get().length()>0){
//					String item = rsps[i].get();
//					String[] strfields = item.split(",");
//					if (strfields.length != FLOWFIELDSIZE) {
//						logger.error("flow field size[" + strfields.length+ "] error, shoule be 19.");
//						continue;
//					}
//					Flow flow = new Flow();
//					flow.num_oldFlow = new long[FLOWFIELDSIZE];
//					long[] num_flow = flow.num_oldFlow;
//	
//					for (int n = 0; n < FLOWFIELDSIZE; n++) {
//						num_flow[n] = Long.parseLong(strfields[n].trim());
//					}
//					
////					esagent.atomProcess(flow);
//	//				esindexer.passiveAtomProcess(flow);
//				}else{
////					esagent.processBulkIfNeeded(true);
//					try {
//						logger.error("no record sleep 5 seconds...");
//						Thread.sleep(5000);
//					} catch (InterruptedException e) {
//						// TODO Auto-generated catch block
//						e.printStackTrace();
//						logger.error(e.getMessage());
//					}
//					break;
//	//				esindexer.processBulkIfNeeded(true);
//				}
//			}
//		}
//		if(jedis != null) {
//			jedisPool.returnResource(jedis);
//		}
//	}
	
//	public static void main(String[] args){
//		JedisBulkGet redisgetter = new JedisBulkGet();
//		redisgetter.bulkGets();
//	}

	public void run() {
		// TODO Auto-generated method stub
//		bulkGets();
	    ScheduledExecutorService checker = Executors.newSingleThreadScheduledExecutor();
	    checker.scheduleAtFixedRate(objPool, 1, 2, TimeUnit.SECONDS);
		while(true){
			loopBulk();
		}
	}
	
	
//	private List<Object> getDataFromRedio() {
//		long t = jedis.llen(redis_key);
//		int length =(int) (t < bulksize ? t : bulksize);
//		for(int i = 0; i < length; i++){
//			jpline.rpop(redis_key);
//		}
//		return jpline.syncAndReturnAll();
//	}
	

	public int loopBulk(){
		Response<String>[] rsps =  new Response[config.getRedisoncedealnum()];	
		for (int i = 0; i < redisoncesize; i++) {
			rsps[i] = (Response<String>) jpline.rpop(redisKey);
		}
		try {
			jpline.sync();
		} catch (Exception e) {
			logger.error("!!!!------------redis sync get rsp fail: "+ e.getMessage());
			return -1;
		}
		for (int i = 0; i < redisoncesize; i++) {
			if (rsps[i] == null || rsps[i].get() == null) {
				break;
			}
			dealRecord(rsps[i].get());
		}
		
		cycle++;
		return redisoncesize;
	}
	
	public void dealRecord(String item) {
		String[] strfields = item.split(",");
		if (strfields.length != FLOWFIELDSIZE) {
			logger.error("flow field size[" + strfields.length+ "] error, shoule be 19.");
			return;
		}
		Flow flow = new Flow();
		flow.num_oldFlow = new long[FLOWFIELDSIZE];
		long[] num_flow = flow.num_oldFlow;

		for (int n = 0; n < FLOWFIELDSIZE; n++) {
			num_flow[n] = Long.parseLong(strfields[n].trim());
		}
		augmentmaker.augment(flow);
		
		if (first) {
			record_starttime = num_flow[0];
			last_time = System.currentTimeMillis();
			first = false;
		}

		record_time = num_flow[0];
		now_time = System.currentTimeMillis();
		
		/*前提条件获取汇聚一个session的时间 < interval时间*/
//		if ((record_time - record_starttime) >= aggr_interval) {
//			logger.info("up:"+(record_time-record_starttime)+", down:"+(now_time-last_time)+", division:"+((record_time-record_starttime) / (now_time-last_time)));
//			if((record_time-record_starttime) / (now_time-last_time)>1.2){
//				sleepSec = (sleepSec*2<MAXSLEEP ? sleepSec*2 : MAXSLEEP);
//			}else if((record_time-record_starttime) / (now_time-last_time+sleepSec*1000)<1.4){
//				sleepSec = (sleepSec/2>MINSLEEP ? sleepSec/2 : MINSLEEP);
//			}
//			logger.info("record time occur Index, RecordTime:"+ record_time+",  doc:" + docounter+", sleep:"+sleepSec);
//			execIndex();
//		}else if((now_time-last_time) >= aggr_interval*1000){
//			logger.info("real time occur Index, RecordTime:"+ record_time+",  doc:" + docounter + ", sleep:"+sleepSec);
//			execIndex();
//		}
		if ((record_time - record_starttime) >= aggr_interval) {
//			logger.info("up:"+(record_time-record_starttime)+", down:"+(now_time-last_time)+", division:"+((record_time-record_starttime)*1000 / (now_time-last_time)));
			if(now_time == last_time);
			else if((record_time-record_starttime)*1000 / (now_time-last_time)>1.2){
				sleepSec = (sleepSec*1.5<MAXSLEEP ? (int)(sleepSec*1.5) : MAXSLEEP);
			}else if((record_time-record_starttime)*1000 / (now_time-last_time)<1.4){
				sleepSec = (sleepSec/1.5>MINSLEEP ? (int)(sleepSec/1.5) : MINSLEEP);
			}
			logger.info("record time occur Index, RecordTime:"+ record_time+", doc:" + docounter+", total doc:"+totalcounter+", sleep:"+sleepSec);
			execIndex();
		}else if((now_time-last_time) >= aggr_interval*1000){
			logger.info("real time occur Index, RecordTime:"+ record_time+", doc:" + docounter+", total doc:"+totalcounter+", sleep:"+sleepSec);
			execIndex();
		}
//		else if(docounter%1000==0){
//			logger.info("cycle:"+ cycle+",  doc:" + docounter);
//		}
		
		aggrate.push(flow);
//		needexec = true;
		docounter++;
		
		totalcounter++;
//		if(totalcounter%1000000==0){
//			logger.info("aggregrate statistic, cycle:" +cycle+ ",  total doc:"+totalcounter);
//		}
	}
	
	
	private void execIndex(){
//		if(!needexec){
//			return;
//		}
//		PassitiveAgent pagentthread = new PassitiveAgent(config.getPesIndexName(),config.getPesIndexType());
		PassitiveAgent pagentthread = objPool.getObject();
		pagentthread.setFlowIterator(aggrate.getAggrMapIterator());
		
		try{
			scheduler.execute(pagentthread);
		}catch(RejectedExecutionException e){
			logger.error("newCachedThreadPool Reject: "+e.getMessage());
		}catch(Exception e){
			logger.error("newCachedThreadPool Exception: "+e.getMessage());
		}
		
		
		try {
			Thread.sleep(sleepSec*1000);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			logger.error(e.getMessage());
		}
		
		aggrate = new Aggregate();
//		record_starttime = record_time;
//		last_time = now_time + sleepSec*1000;
		first = true;
		docounter = 0;
//		needexec = false;
	}
	
	public static void setRunfg(boolean runfg) {
		PassitiveRedisRiver.runfg = runfg;
	}


}
